package com.feidee.fd.sml.algorithm.component.feature

import org.apache.spark.ml.PipelineStage
import org.apache.spark.ml.feature.StopWordsRemover
import org.apache.spark.sql.DataFrame

/**
  * @Author JunxinWang
  * @Date 2019/3/25 19:52
  * @Description 过滤器
  * @Reviewer YongChen
  */
case class StopWordsRemoveEncoderParam(
                                        override val input_pt: String,
                                        override val output_pt: String,
                                        override val hive_table: String,
                                        override val flow_time: String,
                                        override val inputCol: String,
                                        override val outputCol: String,
                                        override val preserveCols: String,
                                        override val modelPath: String,
                                        // 是否区分大小写。默认值为：false
                                        caseSensitive: Boolean,
                                        // 要过滤的词组
                                        stopWords: Array[String]
                                      ) extends FeatureParam {

  def this() = this(null, null, null, null, "input", "features", null, null,false, Array.empty[String])

  override def verify(): Unit = {
    super.verify()
    require(stopWords.nonEmpty, "stopWords Array should not be empty!")
  }

  override def toMap: Map[String, Any] = {
    var map = super.toMap
    map += ("caseSensitive" -> caseSensitive)
    map += ("stopWords" -> stopWords)
    map
  }
}


class StopWordsRemoveEncoder extends AbstractFeatureEncoder[StopWordsRemoveEncoderParam] {

  override def setUp(param: StopWordsRemoveEncoderParam, data: DataFrame): Array[PipelineStage] = {
    val remover = new StopWordsRemover()
      .setInputCol(param.inputCol)
      .setOutputCol(param.outputCol)
      .setCaseSensitive(param.caseSensitive)

      if(tool.isNotNull(param.stopWords)){
        remover.setStopWords(param.stopWords)
      }
    Array(remover)
  }

}

object StopWordsRemoveEncoder {

  def apply(paramStr: String): Unit = {
    new StopWordsRemoveEncoder()(paramStr)
  }

  def main(args: Array[String]): Unit = {
    StopWordsRemoveEncoder(args(0))
  }
}
